package com.itzx.controller;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.itzx.common.Common;
import com.itzx.common.TaskCallBack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;

import static com.itzx.common.Common.ZOOKEEPER_CONNECT;

/**
 * @Author yyeleven
 */
public class CanalTask implements Runnable {
    private String mIp, mDesc;
    private int mPort;
    private boolean isRunning = true;
    private TaskCallBack mTaskCallBack;
    private Logger mLogger;

    public CanalTask(String ip, int port, String desc, TaskCallBack taskCallBack) {
        this.mDesc = desc;
        this.mIp = ip;
        this.mPort = port;
        this.mTaskCallBack = taskCallBack;
        mLogger = LoggerFactory.getLogger(this.getClass());
    }

    public void run() {
//        CanalConnector mConnector = CanalConnectors.newSingleConnector(
//                new InetSocketAddress(mIp, mPort), mDesc, "", "");
        //      基于zookeeper动态获取canal server的地址，建立链接，其中一台server发生crash，可以支持failover
        CanalConnector mConnector = CanalConnectors.newClusterConnector(ZOOKEEPER_CONNECT, mDesc, "", "");
        Assert.notNull(mConnector, "connector is null");
        while (isRunning) {
            try {
                mConnector.connect();
                mConnector.subscribe();
                while (isRunning) {
                    Message message = mConnector.getWithoutAck(Common.BATCH_SIZE); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (size != 0) {
                        mTaskCallBack.running(message.getEntries());
                    }
                    mConnector.ack(batchId); // 提交确认
                }
            } catch (Exception e) {
                mConnector.rollback();
                mTaskCallBack.failError(e);
                isRunning = false;
                mLogger.error(this.getClass().getSimpleName() + " running task is ", e);
            } finally {
                isRunning = false;
                mConnector.disconnect();
                MDC.remove("destination");
            }
        }
    }
}
